package buaa.qel.rpc.client.handler;

import buaa.qel.rpc.client.asyc.RpcPromise;
import buaa.qel.rpc.client.core.ConnectionManager;
import buaa.qel.rpc.common.info.RegistryNodeInfo;
import buaa.qel.rpc.common.message.Beat;
import buaa.qel.rpc.common.message.RpcRequestMessage;
import buaa.qel.rpc.common.message.RpcResponseMessage;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 对一个地址的请求发送和结果回收由一个handler对象绑定并处理
 */
@Slf4j
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponseMessage> {
    //正在等待回收结果的promise，通过序列号取得promise对象
    private final Map<String, RpcPromise> rpcRequestPromises = new ConcurrentHashMap<>();
    private volatile Channel channel;
    private RegistryNodeInfo registryNodeInfo;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    /**
     * 注册时绑定channel
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        this.channel = ctx.channel();
    }

    /**
     * 在读取到response消息后，从排队的rpcFuture中取出，标记为完成后从队列中删除
     * @param ctx
     * @param response
     * @throws Exception
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, RpcResponseMessage response) {
        String sequenceId = response.getSequenceId();
        log.debug("Receive response: " + sequenceId);
        RpcPromise promise = rpcRequestPromises.get(sequenceId);
        if (promise != null) {
            promise.done(response,promise);
            rpcRequestPromises.remove(sequenceId);
        } else {
            log.warn("Can not get pending response for request id: " + sequenceId);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Client caught exception: " + cause.getMessage());
        ctx.close();
    }

    public void close() {
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 2022.4.22更新弃用
     * 先创建一个future对象，然后加入等待的队列中
     * 然后将请求写出，返回future对象，后面收到response后处理future
     * @param request
     * @return
     */
//    public RpcFuture sendRequest(RpcRequestMessage request) {
//        RpcFuture rpcFuture = new RpcFuture(request);
//        pendingRPC.put(request.getSequenceId(), rpcFuture);
//        try {
//            ChannelFuture channelFuture = channel.writeAndFlush(request).sync();
//            DefaultPromise<Object> promise = new DefaultPromise<>(channel.eventLoop());
//            if (!channelFuture.isSuccess()) {
//                log.error("Send request {} error", request.getSequenceId());
//            }
//        } catch (InterruptedException e) {
//            log.error("Send request exception: " + e.getMessage());
//        }
//        return rpcFuture;
//    }

    /**
     * 不用Future接口实现结果的异步获取，使用netty提供的promise
     * 先指定获取promise结果的线程，添加到等待队列中，再发送rpc请求消息
     * @param request
     * @return
     */
    public RpcPromise sendRequestGetPromise(RpcRequestMessage request) {
        RpcPromise promise = new RpcPromise(channel.eventLoop());
        rpcRequestPromises.put(request.getSequenceId(), promise);
        try {
            ChannelFuture channelFuture = channel.writeAndFlush(request).sync();
            if (!channelFuture.isSuccess()) {
                log.error("Send request {} error", request.getSequenceId());
            }
        } catch (InterruptedException e) {
            log.error("Send request exception: " + e.getMessage());
        }
        return promise;
    }

    /**
     * 发送心跳包，不要加入future等待队列中
     */
    private void sendBeatMessage(){
        try {
            ChannelFuture channelFuture = channel.writeAndFlush(Beat.BEAT_MESSAGE).sync();
            if (!channelFuture.isSuccess()) {
                log.error("Send beat failed");
            }
        } catch (InterruptedException e) {
            log.error("Send beat exception: " + e.getMessage());
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            sendBeatMessage();
            log.debug("Client send beat-ping to " + this.channel.remoteAddress());
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    public void setRegistryNodeInfo(RegistryNodeInfo registryNodeInfo) {
        this.registryNodeInfo = registryNodeInfo;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        ConnectionManager.getInstance().removeHandler(registryNodeInfo);
    }
}
